package com.gzhryc.socket.mosquitto;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.gzhryc.common.logging.Logger;
import com.gzhryc.socket.SocketCallback;
import com.gzhryc.socket.SocketConfig;
import com.gzhryc.socket.SocketEvent;
import com.gzhryc.socket.SocketMessage;
import com.gzhryc.socket.SocketNode;
import com.gzhryc.socket.mosquitto.log.ClientConnectedLogHandler;
import com.gzhryc.socket.mosquitto.log.ClientDisconnectedLogHandler;
import com.gzhryc.socket.mosquitto.log.ClientTimeoutLogHandler;
import com.gzhryc.socket.mosquitto.log.DefaultHandler;
import com.gzhryc.socket.mosquitto.log.SocketErrorLogHandler;

public class MosquittoThreadPoolMqttServer extends SocketNode{

	static Logger log = Logger.getLogger(MosquittoThreadPoolMqttServer.class);

	static ConcurrentHashMap<Long,MosquittoThreadPoolMqttServer> Store = new ConcurrentHashMap<Long,MosquittoThreadPoolMqttServer>();
	
	public static MosquittoThreadPoolMqttServer getMosquittoService(Long key){
		return Store.get(key);
	}
	
	public static void setMosquittoService(Long key,MosquittoThreadPoolMqttServer mosquittoService){
		Store.put(key, mosquittoService);
	}
	
	private ThreadPoolExecutor executor;
	private BlockingQueue<Runnable> blockQueue;
	private MainThread mainThread;
	
	SocketCallback mCallback;
	
	public MosquittoThreadPoolMqttServer(SocketConfig config,SocketCallback callback){
		this.config = config;
		this.mCallback = callback;
		this.blockQueue = new LinkedBlockingQueue<Runnable>();
    	this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
    			,config.getKeepAliveTime(), TimeUnit.MILLISECONDS,this.blockQueue); 
	}

	@Override
	public void start() {
		Process ps = null;
		try {
			log.info("启动mosquitto");
			ps = Runtime.getRuntime().exec("mosquitto -c /etc/mosquitto/mosquitto.conf");
			
			//获取进程的标准输入流  
			final InputStream is1 = ps.getInputStream();   
			//获取进程的错误流  
			final InputStream is2 = ps.getErrorStream();  
			
			new InputWorkThread(this,mCallback,is1).start();
			new InputWorkThread(this,mCallback,is2).start();
			
			final MosquittoThreadPoolMqttServer self = this;
			runTask(new Runnable() {
        		@Override
        		public void run() {
        			SocketEvent event = new SocketEvent(self);
        			mCallback.active(event);
        		}
        	});
			ps.waitFor();
			
			ps.getErrorStream().close();
			ps.getInputStream().close();
			ps.getOutputStream().close();
			
			ps.destroy();   
			log.info("关闭mosquitto");
		} catch (IOException e) {
			log.error(e.getMessage(),e);
		} catch (InterruptedException e) {
			log.error(e.getMessage(),e);
		}
		log.info("mosquitto线程关闭");
	}

	@Override
	public void startService() {
		if(mainThread == null || !mainThread.isAlive()) {
			mainThread = new MainThread(this);
			mainThread.setDaemon(true);  //守护线程
			mainThread.start();
		}
	}
	
	public static class MainThread extends Thread {
		
		final MosquittoThreadPoolMqttServer mServer;
		
		public MainThread(MosquittoThreadPoolMqttServer server) {
			mServer = server;
		}
		
		public void run() {
			mServer.start();
		}
	}
	
	public static class InputWorkThread extends Thread {
		
		final InputStream ins;
		final SocketNode mNode;
		final SocketCallback mCallback;
		
		public InputWorkThread(SocketNode node,SocketCallback callback,InputStream ins) {
			this.mNode = node;
			this.mCallback = callback;
			this.ins = ins;
		}
		
		public void run() {
			ClientTimeoutLogHandler clientTimeoutLogHandler = new ClientTimeoutLogHandler(new DefaultHandler(null,mNode,mCallback),mNode,mCallback);
			ClientDisconnectedLogHandler clientDisconnectedLogHandler = new ClientDisconnectedLogHandler(clientTimeoutLogHandler,mNode,mCallback); 
			SocketErrorLogHandler socketErrorLogHandler = new SocketErrorLogHandler(clientDisconnectedLogHandler,mNode,mCallback);
			ClientConnectedLogHandler clientConnectedLogHandler = new ClientConnectedLogHandler(socketErrorLogHandler,mNode,mCallback);
			
			BufferedReader br1 = new BufferedReader(new InputStreamReader(ins));
			try {
				String line1 = null;
				while ((line1 = br1.readLine()) != null) {
					if (line1 != null) {
						try{
							log.info("MQTT 日志：" + line1);
							clientConnectedLogHandler.execute(line1);
						}catch(Exception e){
							log.error(e.getMessage(),e);
						}
					}
				}
			} catch (IOException e) {
				log.error(e.getMessage(),e);
			} finally {
				try {
					ins.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			log.info("MQTT 日志读线程关闭");
		}
	}

	public void runTask(Runnable runnable) {
    	this.executor.submit(runnable);
    }

	@Override
	public boolean writeAndFlush(SocketMessage message) {
		return false;
	}

	@Override
	public void close() {}

	@Override
	public boolean isConnected() {
		return false;
	}

	@Override
	public String getId() {
		return null;
	}

	@Override
	public void reconnect() {
		// TODO Auto-generated method stub
		
	}
}
